package kafka

import (
	"time"

	"github.com/Shopify/sarama"
)

func NewDelayService(brokers []string, topic []string, groupID string) {
	//kafkaConfig := sarama.NewConfig()
	//kafkaConfig.Producer.Return.Successes = true
	//kafkaConfig.Producer.RequiredAcks = sarama.WaitForAll
	//consumerGroup, err := sarama.NewConsumerGroup(brokers, groupID, kafkaConfig)
	//if err != nil {
	//	fmt.Println(err.Error())
	//}
	//defer consumerGroup.Close()
	//
	//producer, err := GetSyncProducerClient("default")
	//if err != nil {
	//	fmt.Println(err.Error())
	//}

	//var wg sync.WaitGroup
	//wg.Add(1)
	//
	//consumer := &DealConsumer{
	//	producer: producer.Producer,
	//	delay:    time.Second * 10,
	//}
	//
	//go func() {
	//	for {
	//		if err := consumerGroup.Consume(context.Background(), topic, consumer); err != nil {
	//			break
	//		}
	//	}
	//	defer wg.Done()
	//}()
	//wg.Wait()
}

type DealConsumer struct {
	producer sarama.SyncProducer
	delay    time.Duration
}

func (c *DealConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		now := time.Now() // 如果消息已经超时，把消息发送到真实队列
		if now.Sub(message.Timestamp) >= c.delay {
			_, _, err := c.producer.SendMessage(&sarama.ProducerMessage{
				Topic: "RealTopic",
				Key:   sarama.ByteEncoder(message.Key),
				Value: sarama.ByteEncoder(message.Value),
			})
			if err == nil {
				session.MarkMessage(message, "")
				continue
			}
			return nil
		}
		// 否则休眠一秒
		time.Sleep(time.Second)
		return nil
	}
	return nil
}

func (c *DealConsumer) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

func (c *DealConsumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}
